package org.groupg.practice.Processor;

import io.reactivex.rxjava3.core.*;
import io.reactivex.rxjava3.schedulers.Schedulers;

public class RxJavaIndicatorProcessor {

    private FlowableEmitter<IndicatorEvent> emitter;
    private final Flowable<IndicatorEvent> flowable;

    public RxJavaIndicatorProcessor() {
        // 创建被观察者(多生产者安全)
        FlowableOnSubscribe<IndicatorEvent> source = emitter -> this.emitter = emitter;

        this.flowable = Flowable.create(source, BackpressureStrategy.BUFFER)
            .observeOn(Schedulers.single()) // 消费者在单一线程执行
            .share(); // 共享热流

        // 订阅消费者
        this.flowable.subscribe(this::processEvent);
    }

    // 发布事件的方法
    public void publishIndicator(String name, Object data) {
        if (emitter != null) {
            emitter.onNext(new IndicatorEvent(name, data));
        }
    }

    // 消费者处理逻辑
    private void processEvent(IndicatorEvent event) {
        try {
            System.out.println("Processing: " + event.getIndicatorName());
            // 实际处理逻辑和数据库插入...
            Thread.sleep(10); // 模拟处理耗时
        } catch (Exception e) {
            System.err.println("Error processing " + event.getIndicatorName() + ": " + e.getMessage());
        }
    }

    public void shutdown() {
        if (emitter != null) {
            emitter.onComplete();
        }
    }

    // 事件类
    public static class IndicatorEvent {
        private final String indicatorName;
        private final Object indicatorData;

        public IndicatorEvent(String name, Object data) {
            this.indicatorName = name;
            this.indicatorData = data;
        }

        public String getIndicatorName() {
            return indicatorName;
        }

        public Object getIndicatorData() {
            return indicatorData;
        }
    }
}